package com.xxl.job.executor.my.util;

import com.alibaba.fastjson.JSON;
import com.xxl.job.executor.my.entity.BatchStockEntity;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * <h3>xxl-job</h3>
 * <p>ES操作</p>
 *
 * @author : 来朋
 * @date : 2020-12-19 17:32
 **/
@Slf4j
@Service
public class EsUtil {

    @Autowired
    RestHighLevelClient restHighLevelClient;

    @Value("${myConfig.esindexname}")
    private String es_indexname;


    /**
     * 查询ES中符合条件的批次信息
     */
    public List<BatchStockEntity> queryBatchInfo(List<String> batchList, Date fDate) {
        List<BatchStockEntity> list=new ArrayList<BatchStockEntity>();
        try {
            if (batchList!=null&&batchList.size()>0) {
                String strDate = "";
                SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
                strDate = formatter.format(fDate);
                StringBuilder sBuild = new StringBuilder();
                for (String tmp : batchList) {
                    sBuild.append(tmp + " ");
                }
                //构建es查询条件
                BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
                //"CX0000609192168201112 NJ0000636241910201009 KD0000203257302201129"
                boolQueryBuilder
                        .must(QueryBuilders.matchQuery("batchno", sBuild.toString()).operator(Operator.OR))
                        .must(QueryBuilders.matchQuery("nowdate", strDate))
                        ;
                // 构建查询条件（注意：termsQuery 支持多种格式查询，如 boolean、int、double、string 等，这里使用的是 string 的查询）
                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                searchSourceBuilder.query(boolQueryBuilder);
                searchSourceBuilder.version(true);
                //指定要返回的字段
                String[] includFields = {"province", "area", "field", "segment", "feeder", "nowdate", "batchstate", "batchno",
                        "livestock", "dayold", "inqty", "indayold", "outqty", "outdayold"};
                searchSourceBuilder.fetchSource(includFields, null);
                // 创建查询请求对象，将查询对象配置到其中
                SearchRequest searchRequest = new SearchRequest(es_indexname);
                searchRequest.source(searchSourceBuilder);
                searchSourceBuilder.size(500000);
                // 执行查询，然后处理响应结果
                SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                // 根据状态和数据条数验证是否返回了数据
                if (RestStatus.OK.equals(searchResponse.status()) && searchResponse.getHits().getTotalHits().value > 0) {
                    SearchHits hits = searchResponse.getHits();
                    for (SearchHit hit : hits) {
                        // 将 JSON 转换成对象
                        BatchStockEntity batchChangeEntity = JSON.parseObject(hit.getSourceAsString(), BatchStockEntity.class);
                        batchChangeEntity.setDocid(hit.getId());
                        batchChangeEntity.set_version(hit.getVersion());
                        list.add(batchChangeEntity);
                    }
                }
            }
        } catch (IOException e) {
            System.out.println(e.getMessage());
        }
        finally {
            return list;
        }
    }

    /**
     * 更新文档
     * @param data
     * @param id
     * @return
     */
    public Integer updateBatchChange(String id, Map<String, Object> data) {
        UpdateRequest updateRequest = new UpdateRequest(es_indexname,id);
        updateRequest.upsert(data);
        try {
            restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return 1;
    }

    /**
     * 删除批次当前日期以后的所有存栏信息
     * @param strBatchNO
     * @param strDate
     * @throws IOException
     */
    public void deleteByQuery(String strBatchNO,String strDate) throws IOException {
        //构建es查询条件
        BoolQueryBuilder boolQueryBuilder=new BoolQueryBuilder();
        boolQueryBuilder
                .must(QueryBuilders.matchQuery("batchno",strBatchNO))
                .filter().add(QueryBuilders.rangeQuery("nowdate").format("yyyy-MM-dd").gte(strDate));
        DeleteByQueryRequest deleteByQueryRequest=new DeleteByQueryRequest(es_indexname);
        deleteByQueryRequest.setQuery(boolQueryBuilder);
        restHighLevelClient.deleteByQuery(deleteByQueryRequest,RequestOptions.DEFAULT);
    }

    /**
     * 通过批次号和日期删除
     * @param strBatchNO
     * @param strMinDate
     * @throws IOException
     */
    public void deleteByBatch(String strBatchNO,String strMinDate) throws IOException {
        //构建es查询条件
        BoolQueryBuilder boolQueryBuilder=new BoolQueryBuilder();
        boolQueryBuilder
                .must(QueryBuilders.matchQuery("batchno",strBatchNO))
                .must(QueryBuilders.matchQuery("nowdate",strMinDate));
        DeleteByQueryRequest deleteByQueryRequest=new DeleteByQueryRequest(es_indexname);
        deleteByQueryRequest.setQuery(boolQueryBuilder);
        restHighLevelClient.deleteByQuery(deleteByQueryRequest,RequestOptions.DEFAULT);
    }

    /**
     * 批量增加文档
     * @param updateList
     */
    public void bulkInsert(List<Map<String, Object>> updateList) throws IOException {
        if (updateList!=null&&updateList.isEmpty()==false){
            BulkRequest bulkRequest=new BulkRequest();
            updateList.stream().forEach(list->{
                System.out.println(list);
                String strDocID=list.getOrDefault("docid","").toString();
                UpdateRequest updateRequest = null;
                if (strDocID.isEmpty()==true){
                    bulkRequest.add(new IndexRequest(es_indexname).opType(DocWriteRequest.OpType.CREATE).source(list));
                } else{
                    updateRequest = new UpdateRequest(es_indexname,strDocID);
                    updateRequest.upsert(list);
                    bulkRequest.add(updateRequest);
                }
            });
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure =
                            bulkItemResponse.getFailure();
                    System.out.println("批量新增失败：\n"+failure.getMessage()+"\n");
                }
            }
        }
    }

    /**
     * 批量删除
     * @param deleteList
     * @throws IOException
     */
    public void bulkDelete(List<BatchStockEntity> deleteList) throws IOException {
        if (deleteList!=null&&deleteList.isEmpty()==false){
            BulkRequest bulkRequest=new BulkRequest();
            deleteList.stream().forEach(batch->{
                bulkRequest.add(new DeleteRequest(es_indexname,batch.getDocid()));
            });
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure =
                            bulkItemResponse.getFailure();
                    System.out.println("批量删除失败：\n"+failure.getMessage()+"\n");
                }
            }
        }
    }


}
